package org.yiming.tools.mqtt.client.client;

import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.yiming.tools.mqtt.client.utils.ConsoleUtil;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.*;

/**
 * <p>1、发送失败，处理异常</p>
 * <p>{@link MqttException#REASON_CODE_MAX_INFLIGHT}</p>
 * <p>{@link MqttException#REASON_CODE_CLIENT_TIMEOUT}</p>
 * <p>2、处理失去连接。丢掉信息</p>
 *
 * @author renwenpeng
 * 2021/10/11 10:01
 */
@RequiredArgsConstructor
@Slf4j
public class MqttClient {

    private final ExecutorService threadPoolTaskExecutor;

    private MqttConnectOptions connOpts;
    private org.eclipse.paho.client.mqttv3.MqttClient sampleClient = null;

    private volatile long notDeliveryCompleteCount = 0;
    private final Object countLock = new Object();
    private final Object reConnLock = new Object();
    private volatile boolean isReConn = false;

    @Getter
    private final String url;
    @Getter
    private final String username;
    @Getter
    private final String password;
    @Getter
    private final String clientId;
    @Getter
    private final String name;

    public MqttClient(ExecutorService threadPoolTaskExecutor, MqttConfig mqttConfig, String clientId) {
        this.threadPoolTaskExecutor = threadPoolTaskExecutor;
        this.url = mqttConfig.getUrl();
        this.username = mqttConfig.getUsername();
        this.password = mqttConfig.getPassword();
        this.clientId = clientId;
        this.name = String.format("%s-%s", url, clientId);
    }

    /**
     * 事件回调
     */
    private final MqttCallback mqttCallback =  new MqttCallback() {
        @Override
        public void connectionLost(Throwable throwable) {
            log.info("connection lose", throwable);
        }

        @Override
        public void messageArrived(String topic, MqttMessage mqttMessage) {
            ConsoleUtil.printStart();
            printLog(String.format("messageArrived, messageId=%d, topic=%s, content=%s"
                    , mqttMessage.getId(), topic, new String(mqttMessage.getPayload(), StandardCharsets.UTF_8)));
            ConsoleUtil.printStop();
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
            synchronized (countLock) {
                notDeliveryCompleteCount--;
            }
            ConsoleUtil.printStart();
            printLog((String.format("deliveryComplete, messageId=%d, topic=%s"
                    , iMqttDeliveryToken.getResponse().getMessageId(), iMqttDeliveryToken.getTopics()[0])));
            ConsoleUtil.printStop();
        }
    };

    private class Publish implements Callable<Boolean> {
        private final String topic;
        private final MqttMessage message;

        public Publish(String topic, MqttMessage message) {
            this.topic = topic;
            this.message = message;
        }

        /**
         * 循环重连发送直到成功，QOS=1就失去意义
         * @return boolean
         */
        boolean sendUtilSuccessWithReConnLoop(){
            final int messageId = message.getId();
            long l = System.currentTimeMillis();
            while (true) {
                try {
                    if (isReConn) {
                        synchronized (reConnLock) {
                            sampleClient.publish(topic, message);
                        }
                    } else {
                        sampleClient.publish(topic, message);
                    }
                    log.info(String.format("MqttClient publish success, messageId=%d, cost:%d ms, topic:%s, content-size:%d byte"
                            , messageId, System.currentTimeMillis() - l, topic, message.getPayload().length));
                    return true;
                } catch (MqttException mqttException) {
                    log.info(String.format("MqttClient publish failure, messageId=%d, cost:%d ms, topic:%s, content-size:%d byte"
                            , messageId, System.currentTimeMillis() - l, topic, message.getPayload().length), mqttException);
                    // 重新发送
                    if (mqttException.getReasonCode() == MqttException.REASON_CODE_MAX_INFLIGHT
                            || mqttException.getReasonCode() == MqttException.REASON_CODE_CLIENT_TIMEOUT) {
                        int code = mqttException.getReasonCode();
                        String msg = null;
                        switch (code) {
                            case MqttException.REASON_CODE_CLIENT_TIMEOUT: msg = "Timed out waiting for a response from the server"; break;
                            case MqttException.REASON_CODE_MAX_INFLIGHT: msg = "Too many publishes in progress"; break;
                            default: break;
                        }
                        log.warn("MqttClient auto reConn, cause: {}", msg);
                        reConn();
                        continue;
                    }
                }
                return false;
            }
        }

        /**
         * 仅发送一次，失败会导致太多，actualInFlight >= maxInflight
         * @return boolean
         */
        boolean sendOnce(){
            final int messageId = message.getId();
            long l = System.currentTimeMillis();
            try {
                sampleClient.publish(topic, message);
                log.info(String.format("MqttClient publish success, messageId=%d, cost:%d ms, topic:%s, content-size:%d byte"
                        , messageId, System.currentTimeMillis() - l, topic, message.getPayload().length));
                return true;
            } catch (MqttException mqttException) {
                log.info(String.format("MqttClient publish failure, messageId=%d, cost:%d ms, topic:%s, content-size:%d byte"
                        , messageId, System.currentTimeMillis() - l, topic, message.getPayload().length), mqttException);
            }
            return false;
        }

        @Override
        public Boolean call() {
            log.info(String.format("MqttClient deliveryStart, messageId=%d, topic=%s, notDeliveryCompleteCount:%d", message.getId()
                    , topic, notDeliveryCompleteCount));
            synchronized (countLock) {
                notDeliveryCompleteCount++;
            }
           return sendOnce();
        }
    }

    private void config(String username, String password){
        // 创建链接参数
        connOpts = new MqttConnectOptions();
        // 设置连接的用户名
        connOpts.setUserName(username);
        connOpts.setPassword(password.toCharArray());

        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
        // 在重新启动和重新连接时记住状态
        connOpts.setCleanSession(true);
        // 自动重连
        connOpts.setAutomaticReconnect(true);
        connOpts.setKeepAliveInterval(60);
        // 设置最大正在发送消息的数量，默认10
        connOpts.setMaxInflight(200);
    }

    @PostConstruct
    public void init(CountDownLatch countDownLatch) {
        config(username, password);
        // 获取tcp地址，首选内网地址，避免网络问题
        threadPoolTaskExecutor.execute(() -> {
            int count = 1;
            int countMax = 3;
            while (sampleClient == null && count <= countMax) {
                try {
                    conn();
                } catch (MqttSecurityException e) {
                    log.info(String.format("MqttClient conn error, url=%s", this.url), e);
                    break;
                } catch (MqttException e) {
                    sampleClient = null;
                    log.info(String.format("MqttClient conn error, url=%s, count=%d", this.url, count++), e);
                    try {
                        TimeUnit.SECONDS.sleep(3);
                    } catch (InterruptedException interruptedException) {
                        interruptedException.printStackTrace();
                        Thread.currentThread().interrupt();
                    }
                }
            }
            if (countDownLatch != null) {
                countDownLatch.countDown();
            }
        });
    }

    private void conn() throws MqttException {
        // 内存存储
        try (final MemoryPersistence persistence = new MemoryPersistence()) {
            sampleClient = new org.eclipse.paho.client.mqttv3.MqttClient(this.url, clientId, persistence);
        }
        sampleClient.setCallback(mqttCallback);
        // 设置发送/连接超时时间 30秒
        // 同步发送，默认阻塞10个，就报错；不设置默认时间会一直等待发送结束
        // org.eclipse.paho.client.mqttv3.MqttException: Too many publishes in progress
        //    at org.eclipse.paho.client.mqttv3.internal.ClientState.send(ClientState.java:524)
        //    at org.eclipse.paho.client.mqttv3.internal.ClientComms.internalSend(ClientComms.java:161)
        sampleClient.setTimeToWait(5000);
        // 建立连接
        sampleClient.connect(connOpts);
        printLog("MqttClient conn success");
    }

    @PreDestroy
    public void close() throws MqttException {
        if (sampleClient != null && sampleClient.isConnected()) {
            // 断开连接
            sampleClient.disconnect();
            // 关闭客户端
            sampleClient.close();
        }
    }

    public void subscribe(String topic, int qos) throws MqttException {
        if (sampleClient != null && sampleClient.isConnected()) {
            sampleClient.subscribe(topic, qos);
        } else {
            printLog("not connected, can not subscribe");
        }
    }

    public Future<Boolean> publish(String topic, String content, int qos) {
        if (sampleClient != null && sampleClient.isConnected()) {
            // 创建消息
            MqttMessage message = new MqttMessage(content.getBytes());
            // 设置消息的服务质量
            message.setQos(qos);
            // 发布消息
            return threadPoolTaskExecutor.submit(new Publish(topic, message));
        } else if (sampleClient != null && !sampleClient.isConnected()) {
            printLog("MqttClient connection lost");
        }
        return null;
    }

    public void unsubscribe(String topic) throws MqttException {
        if (sampleClient != null && sampleClient.isConnected()) {
            sampleClient.unsubscribe(topic);
        } else if (sampleClient != null && !sampleClient.isConnected()) {
            printLog("MqttClient connection lost");
        }
    }

    void reConn() {
        // 记录旧连接
        org.eclipse.paho.client.mqttv3.MqttClient mqttClientTmp = sampleClient;
        synchronized (reConnLock) {
            // 是否是原来的 sampleClient
            if (sampleClient == null || sampleClient.equals(mqttClientTmp)) {
                long l = System.currentTimeMillis();
                log.info("MqttClient reConn start");
                isReConn = true;
                try {
                    close();
                } catch (MqttException e) {
                    log.info("MqttClient reConn close error", e);
                } finally {
                    // 直到连接成功
                    while (true) {
                        try {
                            conn();
                            log.info(String.format("MqttClient reConn end, cast:%d ms", System.currentTimeMillis() - l));
                            break;
                        } catch (MqttException e) {
                            log.info("MqttClient reConn conn error", e);
                        }
                    }
                    isReConn = false;
                }
            }
        }
    }

    private void printLog(String content) {
        ConsoleUtil.printStart();
        log.info("[{}] - {}"
                , name, content);
        ConsoleUtil.printStop();
    }
}
